use notify::{Config, RecommendedWatcher, Watcher};
use tokio::{
    io::{AsyncBufReadExt, AsyncSeekExt},
    task::JoinHandle,
    time::sleep,
};

use crate::rpc_server::{rpc_server, RpcServerImpl};

pub struct MyAgent {
    /// server的rpc地址
    server_addr: String,
    /// 本节点的ip地址
    ip_port: String,
}

/// new&run
impl MyAgent {
    /// 创建一个新的MyAgent实例
    /// # Arguments
    /// * `server_addr` - server的rpc地址
    /// * `ip_port` - 本节点的rpc_server地址
    /// # Returns
    /// 一个新的MyAgent实例
    /// # Example
    /// ```
    /// let agent = MyAgent::new("http://192.168.252.128:50051".to_string(), "192.168.31.72:50053".to_string());
    /// ```
    pub fn new(server_addr: String, ip_port: String) -> Self {
        MyAgent {
            server_addr,
            ip_port,
        }
    }

    /// 运行agent
    /// # 核心逻辑
    /// 1. 启动rpc服务器，用于接收来自server的同步文件rpc请求
    /// 2. 向server注册
    /// 3. 监控配置文件变化&&上报异常给server
    pub async fn run(&self) -> anyhow::Result<()> {
        // 启动rpc服务器，用于接收来自server的同步文件rpc请求
        let rpc_handler = self.start_rpc_server().await;

        // 向server注册
        let register_handler = self.register().await;

        // 监控配置文件变化&&上报异常给server
        let monitor_config_file_handler = self.monitor_config_file().await;
        // 等待异步任务完成
        tokio::try_join!(register_handler, rpc_handler, monitor_config_file_handler)?;
        Ok(())
    }
}

// 注册、心跳; 启动rpc服务器
impl MyAgent {
    async fn register(&self) -> JoinHandle<()> {
        // 等待rpc_server启动
        sleep(tokio::time::Duration::from_secs(1)).await;

        let server_addr = self.server_addr.clone();
        let ip = self.ip_port.clone();
        let mut client =
            rpc_server::config_monitor_s_client::ConfigMonitorSClient::connect(server_addr.clone())
                .await
                .unwrap();
        // 向server注册
        tokio::spawn(async move {
            // 创建rpc客户端
            println!("server_addr: {}", server_addr);
            let request = rpc_server::RegisterAgentRequest { agent_address: ip };
            let response = client.register_agent(request).await.unwrap();
            println!("response: {:?}", response);
        })
    }

    async fn start_rpc_server(&self) -> JoinHandle<()> {
        let addr = self.ip_port.clone().parse().unwrap();

        // 启动rpc服务器
        tokio::spawn(async move {
            // 创建rpc服务器
            let rpc_server = RpcServerImpl::new();
            tonic::transport::Server::builder()
                .add_service(
                    rpc_server::config_monitor_a_server::ConfigMonitorAServer::new(rpc_server),
                )
                .serve(addr)
                .await
                .unwrap();
        })
    }
}

// 监控配置文件变化，上报异常给server
impl MyAgent {
    const AUDIT_RULE_KEY_PREFIX: &'static str = "key=\"CM_";
    async fn monitor_config_file(&self) -> JoinHandle<()> {
        let server_addr = self.server_addr.clone();
        let ip_port = self.ip_port.clone();
        tokio::spawn(async move {
            // 监控audit.log文件，匹配"type=SYSCALL" && "key=\"CM_"，如果匹配到则上报异常

            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

            // notify监控audit.log文件是否有变化
            let mut watcher = RecommendedWatcher::new(
                move |res| {
                    tx.send(res).unwrap();
                },
                Config::default(),
            )
            .unwrap();
            watcher
                .watch(
                    std::path::Path::new("/var/log/audit/audit.log"),
                    notify::RecursiveMode::Recursive,
                )
                .unwrap();

            let mut file = tokio::fs::File::open("/var/log/audit/audit.log")
                .await
                .unwrap();
            file.seek(tokio::io::SeekFrom::End(0)).await.unwrap();
            let mut reader = tokio::io::BufReader::new(file);
            while let Some(_) = rx.recv().await {
                let mut new_line = String::new();
                while reader.read_line(&mut new_line).await.unwrap() > 0 {
                    if new_line.contains("type=SYSCALL")
                        && new_line.contains(Self::AUDIT_RULE_KEY_PREFIX)
                    {
                        // 上报异常
                        #[cfg(debug_assertions)]
                        println!("new_line: {}", new_line);

                        // todo在这里直接上报异常不太好
                        rpc_server::config_monitor_s_client::ConfigMonitorSClient::connect(
                            server_addr.clone(),
                        )
                        .await
                        .unwrap()
                        .report_exception(rpc_server::ReportExceptionRequest {
                            agent_address: ip_port.clone(),
                            exception_message: new_line.clone(),
                        })
                        .await
                        .unwrap();
                    }
                    new_line.clear();
                }
            }
        })
    }
}
